package test

import (
	"bytes"
	"encoding/json"
	"fmt"
	"gitee.com/zaper/kafka-service/model"
	"gitee.com/zaper/kafka-service/other/test/consumer_test"
	"gitee.com/zaper/kafka-service/other/test/producer_test"
	"gitee.com/zaper/kafka-service/service/metadata"
	kafka_partitions2 "gitee.com/zaper/kafka-service/service/metadata/model"
	"gitee.com/zaper/kafka-service/service/producer"
	"gitee.com/zaper/kafka-service/service/producer/producer_model"
	"gitee.com/zaper/kafka-service/util"
	logger "gitee.com/zaper/private-logger"
	"io"
	"net/http"
	"runtime/debug"
	"sync"
	"time"
)

func CreatTopicTest(topName string, NumPartition, NumReplications int) {
	topicModel := &kafka_partitions2.TopicModel{
		TopName:         topName,
		NumPartition:    NumPartition,
		NumReplications: NumReplications,
	}
	metadata_manager.GetMetadataManager().CreateTopic(topicModel)
}

func DeleteTopicTest(topName string) {
	topicModel := &kafka_partitions2.TopicModel{
		TopName: topName,
	}
	metadata_manager.GetMetadataManager().DeleteTopic(topicModel)
}

func ChangePartitionTest(topName string, num int32) {
	partitions := &kafka_partitions2.PartitionModel{
		TopName:       topName,
		NumPartitions: num,
	}
	metadata_manager.GetMetadataManager().ChangePartitions(partitions)
}

func ClientProCusTest() {
	var sw sync.WaitGroup
	sw.Add(2)
	go func() {
		defer sw.Done()
		producer_test.ExecClientProductMess()
	}()

	go func() {
		defer sw.Done()
		consumer_test.ExecClientConsumerMess()
	}()
	sw.Wait()
}

func RegisterTest(count int) {
	for i := 0; i < count; i++ {
		go func(n int) {
			defer func() {
				e := recover()
				if e != nil {
					logger.Infof("stack is %s", debug.Stack())
				}
			}()
			ticker := time.NewTicker(10 * time.Second)
			defer ticker.Stop()
			messageModel := "{\"process_name\":\"%s_go_%d\",\"process_ip\":\"127.0.0.1\",\"process_port\":\"3333\",\"process_type\":\"%d\",\"date_time\":\"%s\"}"
			for {
				message := fmt.Sprintf(messageModel, "test", n, n, util.GetCurrentTime())
				mess := &model.MessModel{
					Key:   []byte("register"),
					Value: []byte(message),
				}
				select {
				case <-ticker.C:
					register(mess)
					//retryCheck(checkRegister, message, n, 5, 2)
				}
			}
		}(i)
	}
}

func register(registerMess *model.MessModel) {
	// 每个协程模拟一个生产者进程，创建一个kafka生产者，并绑定writer
	agent := producer.NewProducerAgent()
	agent.ProducerBind(&producer_model.ProducerModel{
		Topic: "hot_patch",
	})

	// 进程上报自己的进程信息
	err := agent.ProducerMess(registerMess)
	if err != nil {
		logger.Errorf("ProducerMess fail %s", err)
		return
	}
}

// 每上报一次进行一次检查，避免多次上报大量生产无用消息, 每次检查10 * 2s.
// 如果检查时间内还未注册成功，则返回false，重新上报进程信息
// 如果检查时间内注册成功，则返回true，进程结束
//

func checkRegister(message string, index int) bool {
	var resp *http.Response
	var err error
	defer func() {
		if resp != nil && resp.Body != nil {
			err = resp.Body.Close()
			if err != nil {
				logger.Errorf("Close http body fail, err is %s", err)
			}
		}
	}()
	resp, err = util.NewHttpClientManager().Post("http://127.0.0.1:443/api/v1/client/register/status",
		"application/json", bytes.NewBuffer([]byte(message)))
	if err != nil {
		logger.Error("register fail")
		return false
	}
	return parseResponse(resp, index)
}

func parseResponse(resp *http.Response, index int) bool {
	if resp == nil {
		logger.Error("resp is nil")
		return false
	}
	if resp.Body == nil {
		logger.Error("resp.Body is nil")
		return false
	}
	resultArr, err := io.ReadAll(resp.Body)
	if err != nil {
		logger.Errorf("read resp.Body fail, err is %s", err)
		return false
	}
	response := util.NewHttpNullResponseError()
	err = json.Unmarshal(resultArr, response)
	if err != nil {
		logger.Infof("response Unmarshal fail %s", err)
		return false
	}
	if response.ErrorCode == 0 {
		logger.Infof("%s_go_%d, %s", "test", index, response.ErrorMess)
		return true
	}
	return false
}

func retryCheck(function func(string, int) bool, message string, index int, times, delay int) bool {
	for i := 0; i < times; i++ {
		rt := function(message, index)
		if rt {
			return rt
		}
		time.Sleep(time.Duration(delay) * time.Second)
	}
	return false
}

//func IoProCusTest() {
//	logger.Info("生产消费者系统启动...")
//	topics := []string{"first_topic", "second_topic"}
//	var groupIds = []string{"groupId_1", "groupId_2"}
//
//	ConsumerModel1 := &model.KafkaConsumerModel{
//		StartFunc:     kafka_consumer.ExecFunc1,
//		GroupId:       groupIds[0],
//		Topics:        topics,
//		CustomerCount: 3,
//	}
//	ConsumerModel2 := &model.KafkaConsumerModel{
//		StartFunc:     kafka_consumer.ExecFunc1,
//		GroupId:       groupIds[1],
//		Topics:        topics,
//		CustomerCount: 3,
//	}
//	var sw sync.WaitGroup
//	sw.Add(10)
//	// 消费组1
//	kafka_consumer.ReaderGroup(ConsumerModel1, &sw)
//	// 消费组2
//	kafka_consumer.ReaderGroup(ConsumerModel2, &sw)
//
//	// 消费组就绪校验
//	kafka_rebalance.Check(groupIds)
//
//	producerModel := &model.KafkaProducerModel{
//		Topics:        topics,
//		ProducerCount: 4,
//	}
//	// 生产者
//	kafka_producer.WriterGroup(producerModel, &sw)
//	sw.Wait()
//}
